/**
 * TransferManager.java
 */
package fss.transfer;

import java.io.IOException;
import java.io.Serializable;
import java.rmi.RemoteException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Observable;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import fss.common.communication.DownloadRequest;
import fss.common.communication.Message;
import fss.common.communication.UploadResponse;
import fss.core.config.FssProperties;
import fss.remote.RemoteClient;
import fss.remote.RemoteClientObserver;
import fss.remote.RemoteService;
import fss.remote.RemoteServiceObserver;
import fss.util.FssUtils;
import fss.util.UtcDate;
import fss.util.rmi.RmiUtils;

/**
 * <ol>
 *  <li>Call this constructor</li>
 *  <li>Register its owner (Remote Service or Remote Client)</li>
 *  <li>Invoke its initialize() method</li>
 *  <li>Start the thread</li>
 * </ol>
 * @author thinh
 *
 */
public class TransferManager implements Serializable {
    private static final long serialVersionUID = 1L;
    /**
     * The worker thread for transfer request.
     * <p>
     * This object holds a transfer identifier that is mapped in the 
     * Transfer Manager status to a Transfer object. 
     * 
     * @author thinh
     *
     */
    public class TransferThread extends Observable implements Runnable, Serializable {
        private static final long serialVersionUID = 1L;
        /**
         * The transfer id that this thread manages.
         */
        private TransferId id = null;
        /**
         * The entired file cached.
         */
        private byte[] cached = null;
        /**
         * Total of data downloaded (in bytes) -- the offset.
         */
        private int downloaded = 0;
        /**
         * Default constructor specifying the tranfers
         * home/data directory.
         */
        public TransferThread(Transfer transfer) {
            this.id = transfer.getId();
            this.downloaded = transfer.getOffset();
            try {
                cached = FssUtils.getFileBytes(homedir, id.getFilename());
                updateTransferFilesize(cached.length);
                updateTransferStatus(StatusType.INPROGRESS);
            } catch (IOException e) {
                e.printStackTrace();
                updateTransferStatus(StatusType.ERROR);
            }
        }
        @Override
        public void run() {
            Date start = (new UtcDate()).getDate();
            System.out.println("TransferThread started: " + getTransfer(id));
            
            byte[] buffer = null;
            int chunksize = FssProperties.getInstance().getFileTransferChunksize();
            // how much is left for this transfer
            int filesize = cached.length; 
            
            /*
             * Main Loop:
             * 
             * The thread will continuously attempt to broadcast an updated Transfer
             * whenever the status for the transfer is INPROGRESS. 
             * 
             * If the status changed to something other than INPROGRESS, then this 
             * loop will end and the thread will broadcast the updated transfer. 
             */
            while(getTransfer(id).getStatus() == StatusType.INPROGRESS) {
                // segment the chunk appropriately
                if((filesize - downloaded) > chunksize) {
                    buffer = new byte[chunksize];
                } else {
                    buffer = new byte[filesize - downloaded];
                }
                
                // get the data from the cache.
                for(int i = 0; i < buffer.length; i++) {
                    buffer[i] = cached[downloaded];
                    downloaded++;
                }
                
                // update the transfer object with the new offset
                // and data buffer
                updateTransferOffset(downloaded);
                updateTransferData(buffer);
                updateTransferSequence(); // update that the transfer is the next sequence
                stateChanged();
                
                // check to see if the downloaded size is the file size
                // if so, break out of the loop
                boolean done = (downloaded == filesize);
                
                if(done) break; // break out if done
            } 
            
            updateTransferStatus(StatusType.COMPLETED);
            decreaseTransferCount(id);
            stateChanged();
            removeTransfer(id); // remove the transfer
            Date end = (new UtcDate()).getDate();
            float delta = (end.getTime() - start.getTime()) / 1000f;
            System.out.println("****************** Transfer Time (" + id + "): " + delta + " second(s)");
        }
        /**
         * Register an remote service to this thread.
         * @param service the remote service.
         * @throws RemoteException error registering the service.
         */
        public void register(RemoteService service) throws RemoteException {
            RemoteServiceObserver ob = new RemoteServiceObserver(service);
            if(service != null) {
                addObserver(ob);
            }
        }
        /**
         * Register an remote client to this thread.
         * @param service the remote client.
         * @throws RemoteException error registering the client.
         */
        public void register(RemoteClient client) throws RemoteException {
            RemoteClientObserver ob = new RemoteClientObserver(client);
            if(client != null) {
                addObserver(ob);
            }
        }
        /**
         * Uses locking to update the Transfer Manager status map
         * of Transfer objects.
         */
        private void updateTransferSequence() {
            LOCK.lock();
            try {
                status.get(id).increaseSequence();
                status.get(id).setLastUpdateTimestamp((new UtcDate()).getDate());
            } finally {
                LOCK.unlock();
            }
        }
        /**
         * Uses locking to update the Transfer Manager status map
         * of Transfer objects.
         * @param s the new transfer status.
         */
        private void updateTransferStatus(StatusType s) {
            LOCK.lock();
            try {
                status.get(id).setStatus(s);
                status.get(id).setLastUpdateTimestamp((new UtcDate()).getDate());
            } finally {
                LOCK.unlock();
            }
        }
        /**
         * Uses locking to update the Transfer Manager status map
         * of Transfer objects.
         * @param data the new data byte array.
         */
        private void updateTransferData(byte[] data) {
            LOCK.lock();
            try {
                status.get(id).setData(data);
                status.get(id).setLastUpdateTimestamp((new UtcDate()).getDate());
            } finally {
                LOCK.unlock();
            }
        }
        /**
         * Uses locking to update the Transfer Manager status map
         * of Transfer objects.
         * @param offset the new offset.
         */
        private void updateTransferOffset(int offset) {
            LOCK.lock();
            try {
                status.get(id).setOffset(offset);
                status.get(id).setLastUpdateTimestamp((new UtcDate()).getDate());
            } finally {
                LOCK.unlock();
            }
        }
        /**
         * Uses locking to update the Transfer Manager status map
         * of Transfer objects.
         * @param filesize the new filesize.
         */
        private void updateTransferFilesize(int filesize) {
            LOCK.lock();
            try {
                status.get(id).setFilesize(filesize);
                status.get(id).setLastUpdateTimestamp((new UtcDate()).getDate());
            } finally {
                LOCK.unlock();
            }
        }
        /**
         * Starts the transfer as its own thread.
         */
        public void start() {
            updateTransferStatus(StatusType.INPROGRESS);
            increaseTransferCount(id);
            stateChanged();
            Thread thread = new Thread(this);
            thread.start();
        }
        /**
         * Something changed about the transfer so we update tha
         * manager's status and notify the observer of the transfer.
         */
        private void stateChanged() {
            // get the actual transfer object from the transfer status
            Transfer transfer = status.get(id);
            setChanged();
            notifyObservers(transfer);
        }
    }
    /**
     * The lock for the main transfer map.
     */
    private final Lock LOCK = new ReentrantLock();
    /**
     * The lock for the upload count.
     */
    private final Lock LOCK_UPLOAD_COUNT = new ReentrantLock();
    /**
     * The lock for the download count.
     */
    private final Lock LOCK_DOWNLOAD_COUNT = new ReentrantLock();
    /**
     * A map of all the transfers. 
     * <p>
     * The key is the trasnfer id object. The assumption is that no object 
     * is requesting to transfer identical files at the SAME time.
     */
    private Map<TransferId, Transfer> status = new HashMap<TransferId, Transfer>();
    /**
     * Upload count.
     */
    private int activeUploadCount = 0;
    /**
     * Download count.
     */
    private int activeDownloadCount = 0;
    /**
     * The homedir for this Transfer Manager (the data directory).
     */
    private String homedir = null;
    /**
     * Constructor specifying the home directory.
     * @param ownerId the owner identifier.
     * @param homedir the home directory where data resides.
     */
    public TransferManager(String homedir) {
        this.homedir = homedir;
    }
    /**
     * Get the transfered for the specified identifier.
     * @param transferId the identifier.
     * @return the Transfer object.
     */
    public synchronized Transfer getTransfer(TransferId transferId) {
        return status.get(transferId);
    }
    /**
     * Remove the transfer specified by the transfer identifier.
     * @param transferId the id.
     * @return the transfer object.
     */
    public synchronized Transfer removeTransfer(TransferId transferId) {
        return status.remove(transferId);
    }
    /**
     * Increase the transfer count -- download or uploaded depending 
     * on the type specified in the transfer identifier.
     * @param id the transfer id.
     */
    private void increaseTransferCount(TransferId id) {
        switch(id.getType()) {
            case DOWNLOAD:
                LOCK_DOWNLOAD_COUNT.lock();
                try {
                    activeDownloadCount++;
                } finally {
                    LOCK_DOWNLOAD_COUNT.unlock();
                }
                break;
            case UPLOAD:
                LOCK_UPLOAD_COUNT.lock();
                try {
                    activeUploadCount++;
                } finally {
                    LOCK_UPLOAD_COUNT.unlock();
                }
                break;
        }
    }
    /**
     * Decrease the transfer count -- download or uploaded depending 
     * on the type specified in the transfer identifier.
     * @param id the transfer id.
     */
    private void decreaseTransferCount(TransferId id) {
        switch(id.getType()) {
            case UPLOAD:
                LOCK_UPLOAD_COUNT.lock();
                try {
                    activeUploadCount--;
                } finally {
                    LOCK_UPLOAD_COUNT.unlock();
                }
                break;
            case DOWNLOAD:
                LOCK_DOWNLOAD_COUNT.lock();
                try {
                    activeDownloadCount--;
                } finally {
                    LOCK_DOWNLOAD_COUNT.unlock();
                }
                break;
        }
    }
    /**
     * Pause the transfer.
     */
    public void pause(TransferId id) {
        LOCK.lock();
        try {
            status.get(id).setStatus(StatusType.PAUSED);
        } finally {
            LOCK.unlock();
        }
    }
    /**
     * Resume the transfer.
     */
    public void resume(TransferId id) {
        LOCK.lock();
        try {
            status.get(id).setStatus(StatusType.INPROGRESS);
        } finally {
            LOCK.unlock();
        }
    }
    /**
     * Cancel the transfer.
     */
    public void cancel(TransferId id) {
        LOCK.lock();
        try {
            status.get(id).setStatus(StatusType.CANCELLED);
        } finally {
            LOCK.unlock();
        }
    }
    /**
     * Set an error on the transfer.
     */
    public void error(TransferId id) {
        LOCK.lock();
        try {
            status.get(id).setStatus(StatusType.ERROR);
        } finally {
            LOCK.unlock();
        }
    }
    /**
     * Get the number of active downloads.
     * @return the download count.
     */
    public int getDownloadCount() {
        return activeDownloadCount;
    }
    /**
     * Get the number of active uploads.
     * @return the upload count.
     */
    public int getUploadCount() {
        return activeUploadCount;
    }
    /**
     * Get the transfers.
     * @return the current transfers.
     */
    public Map<TransferId, Transfer> getTransfers() {
        return status;
    }
    /**
     * Add a transfer to the set.
     * @param transfer the transfer
     */
    private synchronized void addToStatus(Transfer transfer) {
        status.put(transfer.getId(), transfer);
    }
    /**
     * Process a message.
     * @param message the message.
     */
    public void process(Message message) {
        if(message instanceof DownloadRequest) {
            System.out.println("TransferManager.processRequest: Transfer Manager received Download Request from " + message.getSource());
            DownloadRequest dl = (DownloadRequest) message;
            
            Transfer transfer = new Transfer(dl.getTransferId());
            transfer.setOffset((int)dl.getOffset());
            transfer.setLastUpdateTimestamp((new UtcDate()).getDate());
            System.out.println("TransferManager.process: " + transfer);
            
            // the manager initially adds the transfer to its map
            addToStatus(transfer);
            
            TransferThread worker = new TransferThread(transfer); 
            try {
                switch(dl.getTransferId().getSourceType()) {
                    case CLIENT:
                        RemoteClient client = (RemoteClient) RmiUtils.getRemoteObject(dl.getSource());
                        // this will throw a Remote Exception if the client is down
                        worker.register(client);
                        break;
                    case SERVICE:
                        throw new RemoteException("A Service cannot initiate a Download Request to the Transfer Manager, TransferId: " + dl.getTransferId());
                }
                worker.start(); // starts the transfer thread
                System.out.println("Transfer Manager started worker thread for transfer: " + transfer);
            } catch(RemoteException e) {
                System.out.println("**** TRANSFER ERROR ***** Transfer Manager CANNOT process Download Request from " + message.getSource() + ": " + e);
            }
        }
        else if(message instanceof UploadResponse) {
            UploadResponse response = (UploadResponse) message;
            
            Transfer transfer = new Transfer(response.getTransferId());
            transfer.setOffset((int)response.getOffset());
            transfer.setLastUpdateTimestamp((new UtcDate()).getDate());
            System.out.println("TransferManager.process: " + transfer);
            
            // the manager initially adds the transfer to its map
            addToStatus(transfer);
            
            TransferThread worker = new TransferThread(transfer);
            try {
                switch(response.getTransferId().getSourceType()) {
                    case CLIENT:
                        throw new RemoteException("A Remote Client cannot initiate an Upload Response to the Transfer Manager, TransferId: " + response.getTransferId());
                    case SERVICE:
                        // In an upload we have to register the SERVICE to the worker thread so that it pushes data to it.
                        RemoteService service = (RemoteService) RmiUtils.getRemoteObject(response.getTransferId().getDestinationId());
                        // this will throw a Remote Exception if the client is down
                        worker.register(service);
                }
                worker.start(); // starts the transfer thread
                System.out.println("Transfer Manager started worker thread for transfer: " + transfer);
            } catch(RemoteException e) {
                System.out.println("**** TRANSFER ERROR ***** Transfer Manager CANNOT process Download Request from " + message.getSource() + ": " + e);
            }
        }
    }
}
